package com.shisan.tingshu.search.service.impl;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.search.Hit;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.toolkit.Assert;
import com.google.common.collect.Lists;

import java.util.*;
import java.util.concurrent.*;
import java.util.function.Consumer;
import java.util.function.Supplier;
import java.util.stream.Collectors;

import com.google.common.hash.BloomFilter;
import com.google.common.hash.Funnel;
import com.google.common.hash.Funnels;
import com.shisan.tingshu.album.client.AlbumInfoFeignClient;
import com.shisan.tingshu.common.constant.RedisConstant;
import com.shisan.tingshu.common.execption.ShisanException;
import com.shisan.tingshu.common.result.Result;
import com.shisan.tingshu.common.util.PinYinUtils;
import com.shisan.tingshu.model.album.AlbumAttributeValue;
import com.shisan.tingshu.model.album.AlbumInfo;
import com.shisan.tingshu.model.album.BaseCategoryView;
import com.shisan.tingshu.model.search.AlbumInfoIndex;
import com.shisan.tingshu.model.search.AttributeValueIndex;
import com.shisan.tingshu.model.search.SuggestIndex;
import com.shisan.tingshu.search.executor.ExpireThreadExecutor;
import com.shisan.tingshu.search.factory.ScheduleTaskThreadPoolFactory;
import com.shisan.tingshu.search.repository.AlbumInfoIndexRepository;
import com.shisan.tingshu.search.repository.SuggestIndexRepository;
import com.shisan.tingshu.search.runnable.RebuildBloomFilterRunnable;
import com.shisan.tingshu.search.service.ItemService;
import com.shisan.tingshu.user.client.UserInfoFeignClient;
import com.shisan.tingshu.vo.album.AlbumStatVo;
import com.shisan.tingshu.vo.search.AlbumInfoIndexVo;
import com.shisan.tingshu.vo.user.UserInfoVo;
import jakarta.annotation.PostConstruct;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.redisson.api.RBloomFilter;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.shisan.cache.service.CacheOpsService;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.elasticsearch.core.suggest.Completion;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;


@Slf4j
@Service
@SuppressWarnings({"unchecked", "rawtypes"})
public class ItemServiceImpl implements ItemService {

    @Autowired
    private AlbumInfoIndexRepository albumInfoIndexRepository;

    @Autowired
    private SuggestIndexRepository suggestIndexRepository;

    @Autowired
    private AlbumInfoFeignClient albumInfoFeignClient;

    @Autowired
    private UserInfoFeignClient userInfoFeignClient;

    @Autowired
    private ThreadPoolExecutor threadPoolExecutor;

    @Autowired
    private ElasticsearchClient elasticsearchClient;

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Autowired
    private RBloomFilter rBloomFilter;

    @Autowired
    private RedissonClient redissonClient;

    @Autowired
    private ItemServiceImpl itemServiceImpl; // 自己注入自己，记得在application.yaml中开启允许循环依赖

    @Autowired
    private CacheOpsService cacheOpsService;

    // 创建异步线程池
    ExecutorService executorService = Executors.newFixedThreadPool(4);

    ThreadLocal<String> reentrantLockTokenThreadLocal = new ThreadLocal<>();

    BloomFilter<Long> longBloomFilter = null;

    /**
     * ScheduleTaskThreadPoolFactory工厂类+一次性延迟任务+嵌套任务本身实现定时布隆重建
     */
    @PostConstruct
    public void initRebuildBloomFilter() {

//        // 创建一个定时任务线程池，核心线程数为2，用于执行定时或周期性任务
//        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2);
//        // 安排一个周期性任务：RebuildBloomFilterRunnable，每隔7天执行一次
//        scheduledExecutorService.scheduleWithFixedDelay(
//                new RebuildBloomFilterRunnable(redissonClient, redisTemplate, itemServiceImpl), // 要执行的周期性任务
//                0, // 首次立即执行
//                7, TimeUnit.DAYS // 在上一次任务执行完成后，固定间隔7天再次执行（FixedDelay策略）
//        );
//        // 安排一个一次性延迟任务：RebuildBloomFilterRunnable，在10秒后执行。与上面的周期性任务不同，这个任务只执行一次
//        // 那么怎么实现每10秒执行一次呢？在RebuildBloomFilterRunnable的run方法中，再调用一次scheduledExecutorService.schedule()方法，实现每10秒执行一次
//        scheduledExecutorService.schedule(
//                new RebuildBloomFilterRunnable(redissonClient, redisTemplate, itemServiceImpl),
//                10, TimeUnit.SECONDS
//        );

        // 从服务启动开始，每隔7天的凌晨两点执行一次
        // 使用ScheduleTaskThreadPoolFactory工厂类，实现定时任务的线程池对象创建
        ScheduleTaskThreadPoolFactory instance = ScheduleTaskThreadPoolFactory.getINSTANCE();
        Long taskFirstTime = instance.diffTime(System.currentTimeMillis()); // 传入当前时间，计算出距离下次执行任务的时间差
        instance.execute(
                new RebuildBloomFilterRunnable(redissonClient, redisTemplate, itemServiceImpl),
                taskFirstTime,
                TimeUnit.MILLISECONDS);
//        instance.execute(new RebuildBloomFilterRunnable(redissonClient, redisTemplate, itemServiceImpl), 20L, TimeUnit.SECONDS); // 测试用
    }

    /**
     * 初始化本地布隆过滤器
     */
//     @PostConstruct // Spring在创建ItemServiceImpl Bean对象的时候，在其生命周期走到初始化前，会调用该方法
//    public void initLocalBloomFilter() {
//        // 创建化布隆过滤器
//        // 创建漏斗（Funnel）
//        // 漏斗（Funnel）：是 Guava 库中用于将对象转换为字节流的接口，布隆过滤器通过它将元素哈希为位数组的位置
//        // Funnels.longFunnel()：是 Guava 提供的内置漏斗实现，专门用于处理 Long 类型，将长整型数值转换为字节流
//        Funnel<Long> longFunnel = Funnels.longFunnel();
//        // BloomFilter.create()：静态工厂方法，用于创建布隆过滤器实例
//        // longFunnel：指定元素类型（Long）的漏斗，用于元素的哈希转换
//        // 1000000：预期插入的元素数量（容量）。布隆过滤器会根据此值和误判率计算所需的位数组大小
//        // 0.01：期望的误判率（假阳性概率），即当元素实际不存在时，布隆过滤器误判为存在的概率。这里设置为 1%
//        longBloomFilter = BloomFilter.create(longFunnel, 1000000, 0.01);
//        // 将元素放入布隆过滤器器
//        List<Long> albumInfoIdList = getAlbumInfoIdList();
//        albumInfoIdList.stream().forEach(albumId -> {
//            longBloomFilter.put(albumId);
//        });
//        log.info("本地布隆初始化完毕，布隆中的元素个数：{}", longBloomFilter.approximateElementCount());
//    }

    /**
     * 查询所有的专辑id集合
     * @return
     */
    @Override
    public List<Long> getAlbumInfoIdList() {

        Result<List<Long>> albumIds = albumInfoFeignClient.getAlbumInfoIdList();
        List<Long> albumIdsData = albumIds.getData();
        if (CollectionUtils.isEmpty(albumIdsData)) {
            throw new ShisanException(201, "应用中不存在专辑id集合");
        }
        return albumIdsData;
    }

    /**
     * 专辑上架（异步线程池优化版）30ms
     *
     * @param albumId
     */
    @Override
    @SneakyThrows // 悄无声息地抛出受检异常（checked exceptions），而无需在方法签名中显式声明 throws 子句或在方法体内使用 try-catch 块处理
    public void albumOnSale(Long albumId)  {

        // 创建一个计数器
        CountDownLatch countDownLatch = new CountDownLatch(4); // 与线程池线程数保持一致

        // 创建文档对象
        AlbumInfoIndex albumInfoIndex = new AlbumInfoIndex();

        long startTime = System.currentTimeMillis();

        // 给文档对象赋值
        // 查询tingshu_album库下的album_info表，需要调用`service-album`微服务，此处只能用同步RPC调用

        // 远程调用`service-album`微服务，获取专辑基本信息和标签信息
        Future<Long> future = executorService.submit(new Callable<Long>() {
            @Override
            public Long call() throws Exception {
                // 专辑基本信息
                Result<AlbumInfo> albumInfoResult = albumInfoFeignClient.getAlbumInfoAndAttrValue(albumId);
                AlbumInfo albumInfoData = albumInfoResult.getData();
                if (albumInfoData == null) {
                    throw new ShisanException(201, "远程调用专辑微服务，获取专辑基本信息失败");
                }
                albumInfoIndex.setId(albumInfoData.getId()); // 专辑id
                albumInfoIndex.setAlbumTitle(albumInfoData.getAlbumTitle()); // 专辑标题
                albumInfoIndex.setAlbumIntro(albumInfoData.getAlbumIntro()); // 专辑简介
                albumInfoIndex.setCoverUrl(albumInfoData.getCoverUrl()); // 专辑封面
                albumInfoIndex.setIncludeTrackCount(albumInfoData.getIncludeTrackCount()); // 专辑包含的声音集数
                albumInfoIndex.setIsFinished(albumInfoData.getIsFinished().toString()); // 专辑是否完结
                albumInfoIndex.setPayType(albumInfoData.getPayType()); // 专辑付费类型（免费、vip免费、付费）
                albumInfoIndex.setCreateTime(new Date()); // 专辑保存到es的时间
                // 专辑标签信息
                List<AttributeValueIndex> attributeValueIndexs = albumInfoData.getAlbumAttributeValueVoList().stream().map(albumAttributeValue -> {
                    AttributeValueIndex attributeValueIndex = new AttributeValueIndex();
                    attributeValueIndex.setAttributeId(albumAttributeValue.getAttributeId()); // 标签id
                    attributeValueIndex.setValueId(albumAttributeValue.getValueId()); // 标签值id
                    return attributeValueIndex;
                }).collect(Collectors.toList());
                albumInfoIndex.setAttributeValueIndexList(Lists.newArrayList(attributeValueIndexs)); // 专辑标签列表

                countDownLatch.countDown(); // 计数器减1

                return albumInfoData.getUserId();
            }
        });

        // 远程调用`service-user`微服务，获取专辑主播信息
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                // 专辑主播信息
                Long userId = null;
                try {
                    userId = future.get();//Future.get() 是一个阻塞方法：它会强制任务 2 的线程暂停执行，直到任务 1（Callable）执行完毕并通过 Future 返回 userId；
                    // 只有当任务 1 执行完、future 有结果后，任务 2 才会继续调用 service-user 获取主播信息。
                    Result<UserInfoVo> albumInfoVoResult = userInfoFeignClient.getUserInfo(userId);
                    UserInfoVo userInfoVoData = albumInfoVoResult.getData();
                    Assert.notNull(userInfoVoData, "远程调用用户微服务，获取专辑主播信息失败");
                    albumInfoIndex.setAnnouncerName(userInfoVoData.getNickname()); // 专辑主播名字
                } catch (Exception e) {
                    throw new RuntimeException(e);
                } finally {
                    countDownLatch.countDown(); // 注意：此处的计数器减1，要放在finally代码块中，防止future.get()方法抛出异常，导致计数器减1失败
                }
            }
        });

        // 远程调用`service-album`微服务，获取专辑分类信息
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                // 专辑分类信息
                Result<BaseCategoryView> baseCategoryViewResult = albumInfoFeignClient.getAlbumCategory(albumId);
                BaseCategoryView baseCategoryViewData = baseCategoryViewResult.getData();
                Assert.notNull(baseCategoryViewData, "远程调用专辑微服务，获取专辑分类信息失败");
                albumInfoIndex.setCategory1Id(baseCategoryViewData.getCategory1Id()); // 专辑一级分类id
                albumInfoIndex.setCategory2Id(baseCategoryViewData.getCategory2Id()); // 专辑二级分类id
                albumInfoIndex.setCategory3Id(baseCategoryViewData.getCategory3Id()); // 专辑三级分类id
                countDownLatch.countDown();
            }
        });

        // 远程调用`service-album`微服务，获取专辑统计信息
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                // 专辑统计信息
                Result<AlbumStatVo> albumStatVoResult = albumInfoFeignClient.getAlbumStat(albumId);
                AlbumStatVo albumStatVoData = albumStatVoResult.getData();
                if (albumStatVoData == null) {
                    throw new ShisanException(201, "远程调用专辑微服务，获取专辑统计信息失败");
                }
                Integer playStatNum = albumStatVoData.getPlayStatNum();
                Integer subscribeStatNum = albumStatVoData.getSubscribeStatNum();
                Integer buyStatNum = albumStatVoData.getBuyStatNum();
                Integer commentStatNum = albumStatVoData.getCommentStatNum();
                albumInfoIndex.setPlayStatNum(playStatNum); // 专辑播放量
                albumInfoIndex.setSubscribeStatNum(subscribeStatNum); // 专辑订阅量
                albumInfoIndex.setBuyStatNum(buyStatNum); // 专辑购买量
                albumInfoIndex.setCommentStatNum(commentStatNum); // 专辑评论量
                // Double hotScore = commentStatNum * 0.1 + subscribeStatNum * 0.2 + buyStatNum * 0.3 + playStatNum * 0.4;
                // 此处使用一个简单的算法计算热度值
                Double hotScore = new Random().nextDouble();
                albumInfoIndex.setHotScore(hotScore); // 专辑热度值
                countDownLatch.countDown();
            }
        });

        countDownLatch.await(); // 当计数器为0时，才会继续往下执行

        long endTime = System.currentTimeMillis();
        log.info("专辑上架，耗时：{}毫秒", endTime - startTime);

        // 将文档对象保存到es中
        albumInfoIndexRepository.save(albumInfoIndex);

        // 向suggestInfo索引库中保存数据
        SuggestIndex suggestIndex = new SuggestIndex();
        suggestIndex.setId(albumInfoIndex.getId().toString());
        suggestIndex.setTitle(albumInfoIndex.getAlbumTitle());
        // 例：当用户输入“我喜欢纯音乐”时，会返回“我喜欢纯音乐”、“woxihuhancunyinyue”、“wxhcyy”这三个结果
        suggestIndex.setKeyword(new Completion(new String[]{albumInfoIndex.getAlbumTitle()})); // 我喜欢纯音乐
        suggestIndex.setKeywordPinyin(new Completion(new String[]{PinYinUtils.toHanyuPinyin(albumInfoIndex.getAlbumTitle())})); // woxihuhancunyinyue
        suggestIndex.setKeywordSequence(new Completion(new String[]{PinYinUtils.getFirstLetter(albumInfoIndex.getAlbumTitle())})); // wxhcyy
        suggestIndexRepository.save(suggestIndex);
        log.info("保存搜索关键字索引成功");
    }

    /**
     * 专辑下架
     * @param albumId
     */
    @Override
    public void albumOffSale(Long albumId) {
        try {
            albumInfoIndexRepository.deleteById(albumId);
            log.info("专辑下架成功，专辑id：{}", albumId);
        } catch (Exception e) {
            log.error("专辑下架失败，专辑id：{}", albumId);
        }
    }

    /**
     * 专辑批量下架
     */
    @Override
    public void batchAlbumOffSale() {
        try {
            albumInfoIndexRepository.deleteAll();
            log.info("专辑批量下架成功");
        } catch (Exception e) {
            log.error("专辑批量下架失败");
        }
    }

    // 创建一个本地缓存
    HashMap<Long, Map<String, Object>> localCache = new HashMap<>();

    /**
     * 根据专辑id查询专辑详情（仅使用异步）
     * @param albumId
     * @return
     */
//    @Override
//    public Map<String, Object> getAlbumInfo(Long albumId) {
//
//        Long startTime = System.currentTimeMillis();
//        Map<String, Object> albumInfoFromDb = getAlbumInfoFromDb(albumId);
//        Long endTime = System.currentTimeMillis();
//        log.info("仅使用异步，耗时：{}毫秒", endTime - startTime);
//        return albumInfoFromDb;
//    }

    /**
     * 根据专辑id查询专辑详情（使用本地缓存）
     * @param albumId
     * @return
     */
//    @Override
//    public Map<String, Object> getAlbumInfo(Long albumId) {
//
//        Long startTime = System.currentTimeMillis();
//
//        // 查询本地缓存
//        boolean b = localCache.containsKey(albumId);
//        // 本地缓存中存在，直接返回
//        if (b) {
//            log.info("本地缓存命中，耗时：{}毫秒", System.currentTimeMillis() - startTime);
//            return localCache.get(albumId);
//        }
//        // 本地缓存中不存在，查询数据库，然后保存到本地缓存中
//        Map<String, Object> albumInfoFromDb = getAlbumInfoFromDb(albumId);
//        localCache.put(albumId, albumInfoFromDb);
//
//        Long endTime = System.currentTimeMillis();
//        log.info("本地缓存未命中，耗时：{}毫秒", endTime - startTime);
//
//        // 同时将查询到的数据返回给前端
//        return albumInfoFromDb;
//    }

    /**
     * 根据专辑id查询专辑详情（使用分布式缓存Redis）
     * @param albumId
     * @return
     */
//    @Override
//    public Map<String, Object> getAlbumInfo(Long albumId) {
//
//        Long startTime = System.currentTimeMillis();
//        // 查询Redis分布式缓存
//        String s = redisTemplate.opsForValue().get(albumId.toString());
//        // 判断分布式缓存是否命中
//        if (!StringUtils.isEmpty(s)) {
//            log.info("Redis分布式缓存命中，耗时：{}毫秒", System.currentTimeMillis() - startTime);
//            return JSONObject.parseObject(s, Map.class);
//        }
//
//        // Redis分布式缓存未命中，回源查询数据库
//        Map<String, Object> albumInfoFromDb = getAlbumInfoFromDb(albumId);
//        // 将数据库的数据同步到分布式缓存Redis中
//        redisTemplate.opsForValue().set(String.valueOf(albumId), JSONObject.toJSONString(albumInfoFromDb));
//        Long endTime = System.currentTimeMillis();
//        log.info("Redis分布式缓存未命中，耗时：{}毫秒", endTime - startTime);
//        return albumInfoFromDb;
//    }

    /**
     * 根据专辑id查询专辑详情（使用双缓存架构：本地Map缓存+分布式缓存Redis）
     * @param albumId
     * @return
     */
//    @Override
//    public Map<String, Object> getAlbumInfo(Long albumId) {
//
//        Map<String, Object> result = new HashMap<>();
//        Long startTime = System.currentTimeMillis();
//        //  查询一级缓存，即本地Map缓存
//        boolean b = localCache.containsKey(albumId);
//        if (b) {
//            log.info("一级缓存，即本地Map缓存命中，耗时：{}毫秒", System.currentTimeMillis() - startTime);
//            return localCache.get(albumId);
//        }
//        log.info("一级缓存，即本地Map缓存未命中，查询二级缓存，即分布式缓存Redis");
//        // 查询二级缓存，即分布式缓存Redis
//        String s = redisTemplate.opsForValue().get(albumId.toString());
//        if (!StringUtils.isEmpty(s)) { // 分布式缓存命中
//            // 将Redis的数据反序列化成为Map对象
//            result = JSONObject.parseObject(s, Map.class);
//            // 将数据同步到一级缓存
//            localCache.put(albumId, result);
//            log.info("二级缓存，即分布式缓存Redis命中，耗时：{}毫秒", System.currentTimeMillis() - startTime);
//            return result;
//        } else { // 分布式缓存未命中
//            result = getAlbumInfoFromDb(albumId);
//            // 将数据库查询的数据同步到二级缓存Redis和一级缓存本地缓存Map
//            redisTemplate.opsForValue().set(String.valueOf(albumId), JSONObject.toJSONString(result));
//            localCache.put(albumId, result);
//            log.info("二级缓存，即分布式缓存Redis未命中，耗时：{}毫秒", System.currentTimeMillis() - startTime);
//            return result;
//        }
//    }

    /**
     * 根据专辑id查询专辑详情
     * @param albumId
     * @return
     */
    @Override
    public Map<String, Object> getAlbumInfo(Long albumId) {

        // 如果只是调用对缓存进行读写操作的API，即只调用`cache-starter`依赖中的`CacheOpsServiceImpl`方法
//        return getDistroCacheAndLockFinallyRedissonVersion2(albumId);

        // 回源查询数据库
        return getAlbumInfoFromDb(albumId);
    }

    /**
     * V1：分布式缓存Redis+Redis版本的分布式锁，以解决缓存击穿问题
     * @param albumId
     * @return
     */
    @Nullable
    private Map opsDistroCacheAndLockV1(Long albumId) {

        String cacheKey = RedisConstant.CACHE_INFO_PREFIX + albumId; // 缓存key
        String lockKey = RedisConstant.ALBUM_LOCK_SUFFIX + albumId; // 分布式锁key
        // 查询分布式缓存
        String resultFromCache = redisTemplate.opsForValue().get(cacheKey);
        // 判断缓存是否存在
        if (!StringUtils.isEmpty(resultFromCache)) {
            return JSONObject.parseObject(resultFromCache, Map.class);
        }
        // 缓存未命中，则回源查询数据库
        // 加分布式锁
        Boolean aBoolean = redisTemplate.opsForValue().setIfAbsent(lockKey, "lock");
        if (aBoolean) { // 若抢得到锁（即加锁成功）
            Map<String, Object> albumInfoFromDb;
            // 回源查询数据库
            albumInfoFromDb = getAlbumInfoFromDb(albumId);
            // 将数据库查询的数据同步到Redis
            redisTemplate.opsForValue().set(cacheKey, JSONObject.toJSONString(albumInfoFromDb));
            // 释放分布式锁
            redisTemplate.delete(lockKey);
            // 返回数据给前端
            return albumInfoFromDb;
        } else { // 若未抢到锁（即加锁失败）
            try {
                Thread.sleep(200); // 休眠200毫秒，然后重新获取数据（此时上面先抢到锁的请求已经把数据放入了Redis）。这个时间要通过压测来获取
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            String resultFromCacheStr = redisTemplate.opsForValue().get(cacheKey);
            return JSONObject.parseObject(resultFromCacheStr, Map.class);
        }
    }

    /**
     * V2：在极端情况下，抢到锁的线程刚要执行业务，服务器突然宕机。此时Redis中的锁没有被释放，此时会导致死锁发生
     * 解决：
     * 1、客户端主动删除锁：redisTemplate.delete(lockKey);
     * 2、Redis服务端给锁的key设置过期时间：redisTemplate.expire(lockKey, 30, TimeUnit.SECONDS);
     * 底层实现：Redis将设置key和给这个key设置过期时间的操作用一个lua脚本包装起来，然后用一个Redis连接执行这个lua脚本，从而保证这两个动作的原子性
     * @param albumId
     * @return
     */
    @Nullable
    private Map getDistroCacheAndLockV2(Long albumId) {

        String cacheKey = RedisConstant.CACHE_INFO_PREFIX + albumId;
        String lockKey = RedisConstant.ALBUM_LOCK_SUFFIX + albumId;
        // 查询分布式缓存
        String resultFromCache = redisTemplate.opsForValue().get(cacheKey);
        // 判断缓存是否存在
        if (!StringUtils.isEmpty(resultFromCache)) {
            return JSONObject.parseObject(resultFromCache, Map.class);
        }
        // 缓存未命中，则回源查询数据库
        // 加分布式锁。此时给锁传入一个过期时间，防止死锁发生
        Boolean aBoolean = redisTemplate.opsForValue().setIfAbsent(lockKey, "lock", 30, TimeUnit.SECONDS);
        if (aBoolean) { // 若抢得到锁（即加锁成功）
            Map<String, Object> albumInfoFromDb;
            try {
                // 回源查询数据库
                albumInfoFromDb = getAlbumInfoFromDb(albumId);
                // 将数据库查询的数据同步给缓存Redis
                redisTemplate.opsForValue().set(cacheKey, JSONObject.toJSONString(albumInfoFromDb));
            } finally {
                // 释放分布式锁
                redisTemplate.delete(lockKey);
            }
            // 返回数据给前端
            return albumInfoFromDb;
        } else { // 若未抢到锁（即加锁失败）
            try {
                Thread.sleep(200);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            String resultFromCacheStr = redisTemplate.opsForValue().get(cacheKey);
            return JSONObject.parseObject(resultFromCacheStr, Map.class);
        }
    }

    /**
     * v3：在极端情况下，抢到锁的线程1业务的执行时间超过了锁的过期时间，此时会导致锁提前释放，从而会导致其他线程（线程2）抢到锁，然后线程1执行完业务后，释放锁，此时释放的是线程2的锁，而不是线程1的锁，这就是锁的误删问题
     * 原因：锁和释放锁不是原子操作
     * 解决思路：每个线程加锁的时候都给一个锁标识，然后在释放锁的时候，先判断一下这个锁是不是自己加的，如果是则删除，反之则不删除
     * 解决方法：自定义一个lua脚本，该脚本来判断加锁和释放锁是否是同一个线程
     * @param albumId
     * @return
     */
    @Nullable
    private Map getDistroCacheAndLockV3(Long albumId) {

        String cacheKey = RedisConstant.CACHE_INFO_PREFIX + albumId;
        String lockKey = RedisConstant.ALBUM_LOCK_SUFFIX + albumId;
        String token = UUID.randomUUID().toString().replace("-", ""); // 生成一个随机字符串，作为锁的标识
        // 查询分布式缓存
        String resultFromCache = redisTemplate.opsForValue().get(cacheKey);
        // 判断缓存是否存在
        if (!StringUtils.isEmpty(resultFromCache)) {
            return JSONObject.parseObject(resultFromCache, Map.class);
        }
        // 缓存未命中，则回源查询数据库
        // 加分布式锁。此时给锁传入一个过期时间，防止死锁发生
        Boolean aBoolean = redisTemplate.opsForValue().setIfAbsent(lockKey, token, 30, TimeUnit.SECONDS);
        if (aBoolean) { // 若抢得到锁（即加锁成功）
            Map<String, Object> albumInfoFromDb;
            try {
                // 回源查询数据库
                albumInfoFromDb = getAlbumInfoFromDb(albumId);
                // 将数据库查询的数据同步给缓存Redis
                redisTemplate.opsForValue().set(cacheKey, JSONObject.toJSONString(albumInfoFromDb));
            } finally {
                // 释放分布式锁
                // 判断一下这个锁是不是自己加的，如果是则删除，反之则不删除
                String lockValueFromCache = redisTemplate.opsForValue().get(lockKey); // 获取Redis中的锁
                // 自定义 Lua 脚本。该脚本的意思是：如果 Redis 中键为KEYS[1]的值等于参数ARGV[1]，则删除该键并返回 1，否则直接返回 0
                String luaScript = "IF redis.call('get',KEYS[1])==ARGV[1] THEN return redis.call('del',KEYS[1]) ELSE return 0 END";
                // 执行 Lua 脚本
                // new DefaultRedisScript<Long>(luaScript, Long.class)：封装了之前的 Lua 脚本，并指定返回值类型为Long（对应 Lua 脚本中的return 0或return 1）
                // Arrays.asList(lockKey)：对应 Lua 脚本中的KEYS[1]，即要操作的 Redis 键（锁的名称）
                // token对应 Lua 脚本中的ARGV[1]，即锁的唯一标识
                Long execute = redisTemplate.execute(new DefaultRedisScript<Long>(luaScript, Long.class), Arrays.asList(lockKey), token);
                if (execute == 0) {
                    log.error("释放锁失败");
                } else {
                    log.info("释放锁成功");
                }
                // 如果没有上面的lua脚本。有这么一种情况：
                // 执行完业务，锁还未过期
                // 正要释放锁的时候，锁刚好过期，此时其他线程刚好抢到了锁，释放的就是其他线程的锁，一样会出现锁的误删问题
                if (token.equals(lockValueFromCache)) {
                    redisTemplate.delete(lockKey);  // 此时其它线程还未抢到锁，直接删除
                }
            }
            // 返回数据给前端
            return albumInfoFromDb;
        } else { // 若未抢到锁（即加锁失败）
            try {
                Thread.sleep(200);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            String resultFromCacheStr = redisTemplate.opsForValue().get(cacheKey);
            return JSONObject.parseObject(resultFromCacheStr, Map.class);
        }
    }

    /**
     * Finally Version：抢到锁的线程在将从数据库中查询到的数据写入Redis时失败，此时锁不会释放，从而会导致其他线程无法获取到锁，从而无法访问数据库
     * @param albumId
     * @return
     * 解决思路：只要抢到锁的线程没有把自己的活干完，这个抢到锁的线程对应的这个锁key就不能释放掉。只有等抢到锁的线程把活干完了或者干活期间出异常，才让这个锁过期
     * 即：活没干完，则给锁key续期；活干完或者干活期间出现异常，不用再给锁key续期（注意：只有抢到锁，才续期，没抢到就别续）
     * 续期：每隔10s将Redis中的锁key设置为30s
     * 解决方法：启动一个线程，该线程负责完成续期任务
     * 方法1：直接new一个Thread线程，让这个线程一直做续期任务，并且让这个线程作为守护线程。最后再利用Thread的中断机制，完成对续期线程的取消
     * 方法2：用线程池（这个线程池有定时或者延时功能）完成续期
     */
    private Map getDistroCacheAndLockFinallyVersion(Long albumId) {


        String cacheKey = RedisConstant.CACHE_INFO_PREFIX + albumId; // 缓存key
        String lockKey = RedisConstant.ALBUM_LOCK_SUFFIX + albumId; // 分布式锁key
        String token = ""; // 锁的标识
        Boolean acquireLockFlag = false; // 获取锁标志
        // 查询分布式缓存
        String resultFromCache = redisTemplate.opsForValue().get(cacheKey);
        // 判断缓存是否存在，若存在，则直接返回给前端
        if (!StringUtils.isEmpty(resultFromCache)) {
            return JSONObject.parseObject(resultFromCache, Map.class);
        }
        // 缓存未命中，则回源查询数据库
        // 从ThreadLocal中获取令牌值（解决递归的线程进来）
        String s = reentrantLockTokenThreadLocal.get();
        // 判断是否是递归进来的线程（即已经持有锁的线程）
        if (!StringUtils.isEmpty(s)) {
            token = s; // 如果是递归线程，使用已有的token
            acquireLockFlag = true; // 如果是递归线程，直接获取锁成功
        } else {
            // 如果是第一次进来的线程，生成新的token
            token = UUID.randomUUID().toString().replace("-", ""); // 生成一个随机字符串，作为锁的标识
            // 加分布式锁
            acquireLockFlag = redisTemplate.opsForValue().setIfAbsent(lockKey, token, 30, TimeUnit.SECONDS);
        }
        if (acquireLockFlag) { // 若抢得到锁（即加锁成功）
            // 创建并启动续期线程（每10秒将锁的过期时间重置为30秒）
            ExpireThreadExecutor expireThreadExecutor = new ExpireThreadExecutor(redisTemplate, albumId);
            expireThreadExecutor.renewal(30l, TimeUnit.SECONDS);
            Map<String, Object> albumInfoFromDb;
            try {
                // 回源查询数据库
                albumInfoFromDb = getAlbumInfoFromDb(albumId);
                // 将数据库查询的数据同步到Redis缓存
                redisTemplate.opsForValue().set(cacheKey, JSONObject.toJSONString(albumInfoFromDb));
            } finally {
                // 释放分布式锁
                // 判断一下这个锁是不是自己加的，如果是则删除，反之则不删除
                // 自定义 Lua 脚本。该脚本的意思是：如果 Redis 中键为KEYS[1]的值等于参数ARGV[1]，则删除该键并返回 1，否则直接返回 0
                String luaScript = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
                // 执行 Lua 脚本
                // new DefaultRedisScript<Long>(luaScript, Long.class)：封装了之前的 Lua 脚本，并指定返回值类型为Long（对应 Lua 脚本中的return 0或return 1）
                // Arrays.asList(lockKey)：对应 Lua 脚本中的KEYS[1]，即要操作的 Redis 键（锁的名称）
                // token对应 Lua 脚本中的ARGV[1]，即锁的唯一标识
                Long execute = redisTemplate.execute(new DefaultRedisScript<Long>(luaScript, Long.class), Arrays.asList(lockKey), token);
                if (execute == 0) {
                    log.error("释放锁失败");
                } else {
                    log.info("释放锁成功");
                }
                // 从ThreadLocal移除令牌，防止内存泄漏
                reentrantLockTokenThreadLocal.remove();
                // 结束续期任务
                expireThreadExecutor.cancelRenewal();
            }
            // 返回数据给前端
            return albumInfoFromDb;
        } else { // 若未抢到锁（即加锁失败）
            try {
                Thread.sleep(200); // 休眠200毫秒，然后重新获取数据（此时上面先抢到锁的请求已经把数据放入了Redis）。这个时间要通过压测来获取
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            // 查询缓存。正常99%的情况下200ms之后的缓存一定是有数据，所以直接返回给前端即可
            String firstCacheStr = redisTemplate.opsForValue().get(cacheKey);
            if (!StringUtils.isEmpty(firstCacheStr)) {
                return JSONObject.parseObject(firstCacheStr, Map.class);
            }
            // 剩余1%极端情况：抢到锁的线程在将数据库中的数据同步到缓存的时候出现了问题，导致缓存没有数据
            while (true) {
                // 再次查询缓存
                String doubleCacheStr = redisTemplate.opsForValue().get(cacheKey);
                if (!StringUtils.isEmpty(doubleCacheStr)) {
                    return JSONObject.parseObject(doubleCacheStr);
                }
                // 尝试获取锁
                Boolean acquireLock = redisTemplate.opsForValue().setIfAbsent(lockKey, token, 30, TimeUnit.SECONDS);
                if (acquireLock) {
                    // 获取锁成功，将token存入ThreadLocal（实现可重入）
                    reentrantLockTokenThreadLocal.set(token);
                    break; // 退出循环
                }
            }
            // 递归调用自身（此时当前线程已持有锁）
            return getAlbumInfo(albumId);
        }
    }

    /**
     * 最终版本+布隆过滤器
     * @param albumId
     * @return
     */
    private Map getDistroCacheAndLockAndBloomFilter(Long albumId) {


        String cacheKey = RedisConstant.CACHE_INFO_PREFIX + albumId; // 缓存key
        String lockKey = RedisConstant.ALBUM_LOCK_SUFFIX + albumId; // 分布式锁key
        String token = ""; // 锁的标识
        Boolean acquireLockFlag = false; // 获取锁标志

        // 查询布隆过滤器（本地）。解决缓存穿透的随机值攻击
//        boolean b = longBloomFilter.mightContain(albumId);
//        if (!b) {
//            log.info("本地布隆过滤器中不存在访问的数据:{}", albumId);
//            return null;
//        }

        // 查询布隆过滤器（分布式）
        boolean bloomContains = rBloomFilter.contains(albumId);
        if (!bloomContains) {
            return null;
        }

        // 查询分布式缓存
        String resultFromCache = redisTemplate.opsForValue().get(cacheKey);
        // 判断缓存是否存在，若存在，则直接返回给前端
        if (!StringUtils.isEmpty(resultFromCache)) {
            return JSONObject.parseObject(resultFromCache, Map.class);
        }
        // 缓存未命中，则回源查询数据库
        // 从ThreadLocal中获取令牌值（解决递归的线程进来）
        String s = reentrantLockTokenThreadLocal.get();
        // 判断是否是递归进来的线程（即已经持有锁的线程）
        if (!StringUtils.isEmpty(s)) {
            token = s; // 如果是递归线程，使用已有的token
            acquireLockFlag = true; // 如果是递归线程，直接获取锁成功
        } else {
            // 如果是第一次进来的线程，生成新的token
            token = UUID.randomUUID().toString().replace("-", ""); // 生成一个随机字符串，作为锁的标识
            // 加分布式锁
            acquireLockFlag = redisTemplate.opsForValue().setIfAbsent(lockKey, token, 30, TimeUnit.SECONDS);
        }
        if (acquireLockFlag) { // 若抢得到锁（即加锁成功）
            // 创建并启动续期线程（每10秒将锁的过期时间重置为30秒）
            ExpireThreadExecutor expireThreadExecutor = new ExpireThreadExecutor(redisTemplate, albumId);
            expireThreadExecutor.renewal(30l, TimeUnit.SECONDS);
            Map<String, Object> albumInfoFromDb;
            try {
                long ttl = 0l; // 数据的过期时间
                // 回源查询数据库
                albumInfoFromDb = getAlbumInfoFromDb(albumId);
                // 设置数据的过期时间
                if (albumInfoFromDb != null && albumInfoFromDb.size() > 0) { // 如果数据库查询的数据不为空，则设置一个较长的过期时间
                    ttl = 60 * 60 * 24 * 7l;
                } else { // 如果数据库查询的数据为空，则设置一个较短的过期时间
                    ttl = 60 * 60 * 2;
                }
                // 将数据库查询的数据同步到Redis缓存，同时设置过期时间
                redisTemplate.opsForValue().set(cacheKey, JSONObject.toJSONString(albumInfoFromDb), ttl, TimeUnit.SECONDS);
            } finally {
                // 释放分布式锁
                // 判断一下这个锁是不是自己加的，如果是则删除，反之则不删除
                // 自定义 Lua 脚本。该脚本的意思是：如果 Redis 中键为KEYS[1]的值等于参数ARGV[1]，则删除该键并返回 1，否则直接返回 0
                String luaScript = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
                // 执行 Lua 脚本
                // new DefaultRedisScript<Long>(luaScript, Long.class)：封装了之前的 Lua 脚本，并指定返回值类型为Long（对应 Lua 脚本中的return 0或return 1）
                // Arrays.asList(lockKey)：对应 Lua 脚本中的KEYS[1]，即要操作的 Redis 键（锁的名称）
                // token对应 Lua 脚本中的ARGV[1]，即锁的唯一标识
                Long execute = redisTemplate.execute(new DefaultRedisScript<Long>(luaScript, Long.class), Arrays.asList(lockKey), token);
                if (execute == 0) {
                    log.error("释放锁失败");
                } else {
                    log.info("释放锁成功");
                }
                // 从ThreadLocal移除令牌，防止内存泄漏
                reentrantLockTokenThreadLocal.remove();
                // 结束续期任务
                expireThreadExecutor.cancelRenewal();
            }
            // 返回数据给前端
            return albumInfoFromDb;
        } else { // 若未抢到锁（即加锁失败）
            try {
                Thread.sleep(200); // 休眠200毫秒，然后重新获取数据（此时上面先抢到锁的请求已经把数据放入了Redis）。这个时间要通过压测来获取
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            // 查询缓存。正常99%的情况下200ms之后的缓存一定是有数据，所以直接返回给前端即可
            String firstCacheStr = redisTemplate.opsForValue().get(cacheKey);
            if (!StringUtils.isEmpty(firstCacheStr)) {
                return JSONObject.parseObject(firstCacheStr, Map.class);
            }
            // 剩余1%极端情况：抢到锁的线程在将数据库中的数据同步到缓存的时候出现了问题，导致缓存没有数据
            while (true) {
                // 再次查询缓存
                String doubleCacheStr = redisTemplate.opsForValue().get(cacheKey);
                if (!StringUtils.isEmpty(doubleCacheStr)) {
                    return JSONObject.parseObject(doubleCacheStr);
                }
                // 尝试获取锁
                Boolean acquireLock = redisTemplate.opsForValue().setIfAbsent(lockKey, token, 30, TimeUnit.SECONDS);
                if (acquireLock) {
                    // 获取锁成功，将token存入ThreadLocal（实现可重入）
                    reentrantLockTokenThreadLocal.set(token);
                    break; // 退出循环
                }
            }
            // 递归调用自身（此时当前线程已持有锁）
            return getAlbumInfo(albumId);
        }
    }

    /**
     * 最最终版本：Redisson分布式布隆过滤器+Redisson分布式锁
     * @param albumId
     * @return
     */
    @SneakyThrows
    private Map getDistroCacheAndLockFinallyRedissonVersion1(Long albumId) {

        // 1.定义缓存key
        String cacheKey = RedisConstant.CACHE_INFO_PREFIX + albumId; // 缓存key
        String lockKey = RedisConstant.ALBUM_LOCK_SUFFIX + albumId; // 分布式锁key
        long ttl = 0l; // 数据的过期时间
        // 2.查询分布式布隆过滤器
        boolean contains = rBloomFilter.contains(albumId);
        if (!contains) {
            return null;
        }
        // 3.查询缓存
        String jsonStrFromRedis = redisTemplate.opsForValue().get(cacheKey);
        // 3.1 缓存命中
        if (!StringUtils.isEmpty(jsonStrFromRedis)) {
            return JSONObject.parseObject(jsonStrFromRedis, Map.class);
        }
        // 3.2 缓存未命中 查询数据库
        // 3.2.1 添加分布式锁
        RLock lock = redissonClient.getLock(lockKey);
        boolean accquireLockFlag = lock.tryLock(); // tryLock：非阻塞、自动续期
        if (accquireLockFlag) { // 抢到锁
            try {
                // 3.2.2 回源查询数据
                Map<String, Object> albumInfoFromDb = getAlbumInfoFromDb(albumId);
                if (albumInfoFromDb != null) { // 如果根据albumId查询到的数据不为空，则设置一个较长的过期时间
                    ttl = 60 * 60 * 24 * 7l;
                } else { // 如果根据albumId查询到的数据为空，则设置一个较短的过期时间
                    ttl = 60 * 60 * 24l;
                }
                // 3.2.3 同步数据到缓存中去
                redisTemplate.opsForValue().set(cacheKey, JSONObject.toJSONString(albumInfoFromDb), ttl, TimeUnit.SECONDS); // 防止缓存穿透的固定值攻击
                return albumInfoFromDb;
            } finally {
                lock.unlock();// 释放锁
            }
        } else { // 没抢到锁。等同步时间之后，查询缓存即可
            Thread.sleep(200);
            String result = redisTemplate.opsForValue().get(cacheKey);
            if (!StringUtils.isEmpty(result)) {
                return JSONObject.parseObject(result, Map.class);
            }
            return getAlbumInfoFromDb(albumId);
        }
    }

    /**
     * 最最终版本：Redisson分布式布隆过滤器+Redisson分布式锁（抽取缓存操作的方法版本）
     * @param albumId
     * @return
     */
    @SneakyThrows
    private Map getDistroCacheAndLockFinallyRedissonVersion2(Long albumId) {

        // 1.定义缓存key
        String cacheKey = RedisConstant.CACHE_INFO_PREFIX + albumId; // 缓存key
        String lockKey = RedisConstant.ALBUM_LOCK_SUFFIX + albumId; // 分布式锁key
        long ttl = 0l; // 数据的过期时间
        // 2.查询分布式布隆过滤器
        boolean contains = rBloomFilter.contains(albumId);
        if (!contains) {
            return null;
        }
        // 3.查询缓存
        Map dataFromCache = cacheOpsService.getDataFromCache(cacheKey, Map.class);
        // 3.1 缓存命中
        if (dataFromCache != null) {
            return dataFromCache;
        }
        // 3.2 缓存未命中 查询数据库
        // 3.2.1 添加分布式锁
        RLock lock = redissonClient.getLock(lockKey);
        boolean accquireLockFlag = lock.tryLock(); // tryLock：非阻塞、自动续期
        if (accquireLockFlag) { // 抢到锁
            try {
                // 3.2.2 回源查询数据
                Map<String, Object> albumInfoFromDb = getAlbumInfoFromDb(albumId);
                // 3.2.3 同步数据到缓存中去
                cacheOpsService.saveDataToCache(cacheKey, albumInfoFromDb);
                return albumInfoFromDb;
            } finally {
                lock.unlock();// 释放锁
            }
        } else { // 没抢到锁。等同步时间之后，查询缓存即可
            Thread.sleep(200);
            Map result = cacheOpsService.getDataFromCache(cacheKey, Map.class);
            if (result != null) {
                return result;
            }
            return getAlbumInfoFromDb(albumId);
        }
    }

    /**
     * 手动布隆重建
     * @return
     */
    @Override
    public Boolean rebuildBloomFilter() {
        // 步骤：删除老布隆的数据 >> 删除老布隆的配置 >> 创建新布隆 >> 初始化新布隆 >> 将数据放到新布隆
        // 但在高并发场景下，第一个线程删除了老布隆的配置但是新布隆还没有创建时，第二个线程进来仍然使用的是老布隆，此时就会报错
        // 优化做法：创建新布隆 >> 初始化新布隆 >> 将数据放到新布隆 >> 删除老布隆的数据 >> 删除老布隆的配置 >> 将新布隆的名字重命名为老布隆的名字（第4、5、6步要做成一个原子操作）

        // 1、创建新布隆
        RBloomFilter<Object> albumIdBloomFilterNew = redissonClient.getBloomFilter("albumIdBloomFilterNew");
        // 2、初始化新布隆
        albumIdBloomFilterNew.tryInit(1000000l, 0.001);
        // 3、将数据放到新布隆
        List<Long> albumInfoIdList = getAlbumInfoIdList();
        for (Long albumId : albumInfoIdList) {
            albumIdBloomFilterNew.add(albumId);
        }
        albumIdBloomFilterNew.add(2000L); // 给新布隆添加一个老布隆不存在的数据，用于测试
        // 用lua脚本保证这三个步骤的原子性：4、删除老布隆的数据；5、删除老布隆的配置；6、将新布隆的名字重命名为老布隆的名字
        String script = " redis.call(\"del\",KEYS[1])" +
                "  redis.call(\"del\",KEYS[2])" +
                // KEYS[1]对应的是下面asList的第一个元素，KEYS[2]对应的是下面asList的第二个元素，以此类推
                "  redis.call(\"rename\",KEYS[3],KEYS[1])" + // 用后者替换前者
                "  redis.call(\"rename\",KEYS[4],KEYS[2]) return 0";
        List<String> asList = Arrays.asList("albumIdBloomFilter", "{albumIdBloomFilter}:config", "albumIdBloomFilterNew", "{albumIdBloomFilterNew}:config");
        Long execute = redisTemplate.execute(new DefaultRedisScript<Long>(script, Long.class), asList);
        if (execute == 0) {
            log.info("老布隆被删除，新布隆上线");
        }
        return execute == 0;
    }

    /**
     * 从数据库根据专辑id查询专辑详情
     * @param albumId
     * @return
     */
    @NotNull
    private Map<String, Object> getAlbumInfoFromDb(Long albumId) {
        // 创建Map对象
        Map<String, Object> map = new HashMap<>();

        CompletableFuture<Void> albumStatCompletableFuture = CompletableFuture.runAsync(new Runnable() {
            @Override
            public void run() {
                System.out.println("查询专辑的统计信息使用的线程：" + Thread.currentThread().getName());
                // 专辑的统计信息
                Result<AlbumStatVo> albumStatResult = albumInfoFeignClient.getAlbumStat(albumId);
                AlbumStatVo albumStatVoData = albumStatResult.getData();
                if (albumStatVoData == null) {
                    throw new ShisanException(201, "远程查询专辑微服务，获取专辑统计信息失败");
                }
                map.put("albumStatVo", albumStatVoData);
            }
        });

        CompletableFuture<Void> viewCompletableFuture = CompletableFuture.runAsync(new Runnable() {
            @Override
            public void run() {
                System.out.println("查询专辑的分类使用的线程：" + Thread.currentThread().getName());
                // 专辑的分类（分类的名字）
                Result<BaseCategoryView> albumCategoryResult = albumInfoFeignClient.getAlbumCategory(albumId);
                BaseCategoryView baseCategoryViewData = albumCategoryResult.getData();
                if (baseCategoryViewData == null) {
                    throw new ShisanException(201, "远程查询专辑微服务，获取专辑分类信息失败");
                }
                map.put("baseCategoryView", baseCategoryViewData);
            }
        });

        CompletableFuture<Long> albumInfoCompletableFuture = CompletableFuture.supplyAsync(new Supplier<Long>() {
            @Override
            public Long get() {
                System.out.println("查询专辑的基本数据使用的线程：" + Thread.currentThread().getName());
                // 专辑基本数据
                Result<AlbumInfo> albumInfoAndAttrValueResult = albumInfoFeignClient.getAlbumInfoAndAttrValue(albumId);
                AlbumInfo albumInfoData = albumInfoAndAttrValueResult.getData();
                if (albumInfoData == null) {
                    throw new ShisanException(201, "远程查询专辑微服务，获取专辑基本信息失败");
                }
                map.put("albumInfo", albumInfoData);

                return albumInfoData.getUserId();
            }
        }, threadPoolExecutor);

        // CompletableFuture<Void> userInfoCompletableFuture = CompletableFuture.runAsync(new Runnable() {
        // @SneakyThrows
        // @Override
        // public void run() {
        //         // 专辑的主播信息
        //         Long userId = albumInfoCompletableFuture.get();
        //         Result<UserInfoVo> userInfoResult = userInfoFeignClient.getUserInfo(userId);
        //         UserInfoVo userInfoData = userInfoResult.getData();
        //         if (userInfoData == null) {
        //             throw new ShisanException(201, "远程查询用户微服务，获取专辑对应主播信息失败");
        //         }
        //         map.put("announcer", userInfoData);
        //     }
        // });

        // 对于上面注释的另一种写法
        CompletableFuture<Void> userInfoCompletableFuture = albumInfoCompletableFuture.thenAcceptAsync(new Consumer<Long>() {
            @Override
            public void accept(Long userId) {
                System.out.println("查询专辑的主播信息使用的线程：" + Thread.currentThread().getName());
                // 专辑的主播信息
                Result<UserInfoVo> userInfoResult = userInfoFeignClient.getUserInfo(userId);
                UserInfoVo userInfoData = userInfoResult.getData();
                if (userInfoData == null) {
                    throw new ShisanException(201, "远程查询用户微服务，获取专辑对应主播信息失败");
                }
                map.put("announcer", userInfoData);
            }
        });
        // 等待所有异步线程执行完毕，Tomcat的线程才能继续执行
        CompletableFuture.allOf(
                albumStatCompletableFuture,
                viewCompletableFuture,
                albumInfoCompletableFuture,
                userInfoCompletableFuture)
                .join();

        return map;
    }

    /**
     * 将ES中的排行榜数据缓存到Redis中
     */
    @SneakyThrows
    @Override
    public void preRankingToCache() {
        // 查询全平台的一级分类id
        Result<List<Long>> c1IdsResult = albumInfoFeignClient.getAllCategory1Id();
        List<Long> c1IdData = c1IdsResult.getData();
        if (CollectionUtils.isEmpty(c1IdData)) {
            throw new ShisanException(201, "远程查询专辑微服务获取一级分类id失败");
        }

        for (Long c1Id : c1IdData) {
            // 五个排行的维度：热度、播放量、订阅量、购买量、评论量
            String[] fiveDimension = {"hotScore", "playStatNum", "subscribeStatNum", "buyStatNum", "commentStatNum"};
            for (String dimension : fiveDimension) {
                // 使用lambda表达式，直接得到查询es返回的结果
                SearchResponse<AlbumInfoIndex> response = elasticsearchClient
                        .search(srb -> srb
                        .index("albuminfo")
                        .query(qb -> qb
                                .term(tqb -> tqb
                                        .field("category1Id")
                                        .value  (c1Id)))
                        .sort(sob -> sob.field(fsb -> fsb.field(dimension)))
                        .size(10), AlbumInfoIndex.class);
                // 将查询到的数据封装到List中
                List<AlbumInfoIndex> albumInfoIndices = new ArrayList<>();
                for (Hit<AlbumInfoIndex> hit : response.hits().hits()) {
                    AlbumInfoIndex albumInfoIndex = hit.source();
                    albumInfoIndices.add(albumInfoIndex);
                }
                // Redis的数据类型有：string、set、zset、hash（大key、小key）、list
                // 用c1Id作为大key，dimension（维度）作为小key，每个小key对应一个List，List中存放的前十的专辑信息
                String bigKey = RedisConstant.RANKING_KEY_PREFIX + c1Id;
                redisTemplate.opsForHash().put(bigKey, dimension, JSONObject.toJSONString(albumInfoIndices));
            }
        }
    }

    /**
     * 查询排行榜
     * @param c1Id
     * @param dimension
     * @return
     */
    @Override
    public List<AlbumInfoIndexVo> findRankingList(Long c1Id, String dimension) {

        // 从Redis中查询排行榜数据
        String bigKey = RedisConstant.RANKING_KEY_PREFIX + c1Id;
        String albumInfoIndexList = (String) redisTemplate.opsForHash().get(bigKey, dimension);
        if (StringUtils.isEmpty(albumInfoIndexList)) { // 没key
            throw new ShisanException(201, "排行榜信息不存在");
        }
        // 将JSON字符串转换为List<AlbumInfoIndex>对象
        List<AlbumInfoIndex> albumInfoIndices = JSONObject.parseArray(albumInfoIndexList, AlbumInfoIndex.class);
        List<AlbumInfoIndexVo> albumInfoIndexVoList = albumInfoIndices.stream().map(albumInfoIndex -> {
            AlbumInfoIndexVo albumInfoIndexVo = new AlbumInfoIndexVo();
            BeanUtils.copyProperties(albumInfoIndex, albumInfoIndexVo);
            return albumInfoIndexVo;
        }).collect(Collectors.toList());

        return albumInfoIndexVoList;
    }
//    /**
//     * 专辑上架（异步线程优化版。C、D、A先执行，然后B再执行）
//     *
//     * @param albumId
//     */
//    @Override
//    public void albumOnSale(Long albumId) throws InterruptedException {
//
//        // 创建文档对象
//        AlbumInfoIndex albumInfoIndex = new AlbumInfoIndex();
//
//        ConcurrentHashMap<String, Long> cMap = new ConcurrentHashMap<>();
//
//        long startTime = System.currentTimeMillis();
//
//        // 给文档对象赋值
//        // 查询tingshu_album库下的album_info表，需要调用`service-album`微服务，此处只能用同步RPC调用
//
//        // 远程调用`service-album`微服务，获取专辑分类信息
//        Thread threadC = new Thread(new Runnable() {
//            @Override
//            public void run() {
//                // 专辑分类信息
//                Result<BaseCategoryView> baseCategoryViewResult = albumInfoFeignClient.getAlbumCategory(albumId);
//                BaseCategoryView baseCategoryViewData = baseCategoryViewResult.getData();
//                Assert.notNull(baseCategoryViewData, "远程调用专辑微服务，获取专辑分类信息失败");
//                albumInfoIndex.setCategory1Id(baseCategoryViewData.getCategory1Id()); // 专辑一级分类id
//                albumInfoIndex.setCategory2Id(baseCategoryViewData.getCategory2Id()); // 专辑二级分类id
//                albumInfoIndex.setCategory3Id(baseCategoryViewData.getCategory3Id()); // 专辑三级分类id
//            }
//        }, "thread-C");
//        threadC.start();
//
//        // 远程调用`service-album`微服务，获取专辑统计信息
//        Thread threadD = new Thread(new Runnable() {
//            @Override
//            public void run() {
//                // 专辑统计信息
//                Result<AlbumStatVo> albumStatVoResult = albumInfoFeignClient.getAlbumStat(albumId);
//                AlbumStatVo albumStatVoData = albumStatVoResult.getData();
//                if (albumStatVoData == null) {
//                    throw new ShisanException(201, "远程调用专辑微服务，获取专辑统计信息失败");
//                }
//                Integer playStatNum = albumStatVoData.getPlayStatNum();
//                Integer subscribeStatNum = albumStatVoData.getSubscribeStatNum();
//                Integer buyStatNum = albumStatVoData.getBuyStatNum();
//                Integer commentStatNum = albumStatVoData.getCommentStatNum();
//                albumInfoIndex.setPlayStatNum(playStatNum); // 专辑播放量
//                albumInfoIndex.setSubscribeStatNum(subscribeStatNum); // 专辑订阅量
//                albumInfoIndex.setBuyStatNum(buyStatNum); // 专辑购买量
//                albumInfoIndex.setCommentStatNum(commentStatNum); // 专辑评论量
//                // Double hotScore = commentStatNum * 0.1 + subscribeStatNum * 0.2 + buyStatNum * 0.3 + playStatNum * 0.4;
//                // 此处使用一个简单的算法计算热度值
//                Double hotScore = new Random().nextDouble();
//                albumInfoIndex.setHotScore(hotScore); // 专辑热度值
//            }
//        }, "thread-D");
//        threadD.start();
//
//        // 远程调用`service-album`微服务，获取专辑基本信息和标签信息
//        Thread threadA = new Thread(new Runnable() {
//            @Override
//            public void run() {
//                // 专辑基本信息
//                Result<AlbumInfo> albumInfoResult = albumInfoFeignClient.getAlbumInfoAndAttrValue(albumId);
//                AlbumInfo albumInfoData = albumInfoResult.getData();
//                if (albumInfoData == null) {
//                    throw new ShisanException(201, "远程调用专辑微服务，获取专辑基本信息失败");
//                }
//                albumInfoIndex.setId(albumInfoData.getId()); // 专辑id
//                albumInfoIndex.setAlbumTitle(albumInfoData.getAlbumTitle()); // 专辑标题
//                albumInfoIndex.setAlbumIntro(albumInfoData.getAlbumIntro()); // 专辑简介
//                albumInfoIndex.setCoverUrl(albumInfoData.getCoverUrl()); // 专辑封面
//                albumInfoIndex.setIncludeTrackCount(albumInfoData.getIncludeTrackCount()); // 专辑包含的声音集数
//                albumInfoIndex.setIsFinished(albumInfoData.getIsFinished().toString()); // 专辑是否完结
//                albumInfoIndex.setPayType(albumInfoData.getPayType()); // 专辑付费类型（免费、vip免费、付费）
//                albumInfoIndex.setCreateTime(new Date()); // 专辑保存到es的时间
//                // 专辑标签信息
//                List<AttributeValueIndex> attributeValueIndexs = albumInfoData.getAlbumAttributeValueVoList().stream().map(albumAttributeValue -> {
//                    AttributeValueIndex attributeValueIndex = new AttributeValueIndex();
//                    attributeValueIndex.setAttributeId(albumAttributeValue.getAttributeId()); // 标签id
//                    attributeValueIndex.setValueId(albumAttributeValue.getValueId()); // 标签值id
//                    return attributeValueIndex;
//                }).collect(Collectors.toList());
//                albumInfoIndex.setAttributeValueIndexList(Lists.newArrayList(attributeValueIndexs)); // 专辑标签列表
//                cMap.put("userId", albumInfoData.getUserId());
//            }
//        }, "thread-A");
//        threadA.start();
//        threadA.join(); // 先阻塞 Tomccat 线程，等待线程A执行完毕，再执行后续代码
//
//        // 远程调用`service-user`微服务，获取专辑主播信息
//        Thread threadB = new Thread(new Runnable() {
//            @Override
//            public void run() {
//                Long userId = cMap.get("userId");
//                // 专辑主播信息
//                Result<UserInfoVo> albumInfoVoResult = userInfoFeignClient.getUserInfo(userId);
//                UserInfoVo userInfoVoData = albumInfoVoResult.getData();
//                Assert.notNull(userInfoVoData, "远程调用用户微服务，获取专辑主播信息失败");
//                albumInfoIndex.setAnnouncerName(userInfoVoData.getNickname()); // 专辑主播名字
//            }
//        }, "thread-B");
//        threadB.start();
//
//        long endTime = System.currentTimeMillis();
//        log.info("专辑上架，耗时：{}毫秒", endTime - startTime);
//
//        // 将文档对象保存到es中
//        albumInfoIndexRepository.save(albumInfoIndex);
//    }

//    /**
//     * 专辑上架
//     *
//     * @param albumId
//     */
//    @Override
//    public void albumOnSale(Long albumId) {
//
//        // 创建文档对象
//        AlbumInfoIndex albumInfoIndex = new AlbumInfoIndex();
//
//        long startTime = System.currentTimeMillis();
//
//        // 给文档对象赋值
//        // 查询tingshu_album库下的album_info表，需要调用`service-album`微服务，此处只能用同步RPC调用
//        Result<AlbumInfo> albumInfoResult = albumInfoFeignClient.getAlbumInfoAndAttrValue(albumId);
//        AlbumInfo albumInfoData = albumInfoResult.getData();
//        if (albumInfoData == null) {
//            throw new ShisanException(201, "远程调用专辑微服务，获取专辑基本信息失败");
//        }
//        // 专辑基本信息
//        albumInfoIndex.setId(albumInfoData.getId()); // 专辑id
//        albumInfoIndex.setAlbumTitle(albumInfoData.getAlbumTitle()); // 专辑标题
//        albumInfoIndex.setAlbumIntro(albumInfoData.getAlbumIntro()); // 专辑简介
//        albumInfoIndex.setCoverUrl(albumInfoData.getCoverUrl()); // 专辑封面
//        albumInfoIndex.setIncludeTrackCount(albumInfoData.getIncludeTrackCount()); // 专辑包含的声音集数
//        albumInfoIndex.setIsFinished(albumInfoData.getIsFinished().toString()); // 专辑是否完结
//        albumInfoIndex.setPayType(albumInfoData.getPayType()); // 专辑付费类型（免费、vip免费、付费）
//        albumInfoIndex.setCreateTime(new Date()); // 专辑保存到es的时间
//        // 专辑标签信息
//        List<AttributeValueIndex> attributeValueIndexs = albumInfoData.getAlbumAttributeValueVoList().stream().map(albumAttributeValue -> {
//            AttributeValueIndex attributeValueIndex = new AttributeValueIndex();
//            attributeValueIndex.setAttributeId(albumAttributeValue.getAttributeId()); // 标签id
//            attributeValueIndex.setValueId(albumAttributeValue.getValueId()); // 标签值id
//            return attributeValueIndex;
//        }).collect(Collectors.toList());
//        albumInfoIndex.setAttributeValueIndexList(Lists.newArrayList(attributeValueIndexs)); // 专辑标签列表
//        // 专辑主播信息
//        Result<UserInfoVo> albumInfoVoResult = userInfoFeignClient.getUserInfo(albumInfoData.getUserId());
//        UserInfoVo userInfoVoData = albumInfoVoResult.getData();
//        Assert.notNull(userInfoVoData, "远程调用用户微服务，获取专辑主播信息失败");
//        albumInfoIndex.setAnnouncerName(userInfoVoData.getNickname()); // 专辑主播名字
//        // 专辑分类信息
//        Result<BaseCategoryView> baseCategoryViewResult = albumInfoFeignClient.getAlbumCategory(albumId);
//        BaseCategoryView baseCategoryViewData = baseCategoryViewResult.getData();
//        Assert.notNull(baseCategoryViewData, "远程调用专辑微服务，获取专辑分类信息失败");
//        albumInfoIndex.setCategory1Id(baseCategoryViewData.getCategory1Id()); // 专辑一级分类id
//        albumInfoIndex.setCategory2Id(baseCategoryViewData.getCategory2Id()); // 专辑二级分类id
//        albumInfoIndex.setCategory3Id(baseCategoryViewData.getCategory3Id()); // 专辑三级分类id
//        // 专辑统计信息
//        Result<AlbumStatVo> albumStatVoResult = albumInfoFeignClient.getAlbumStat(albumId);
//        AlbumStatVo albumStatVoData = albumStatVoResult.getData();
//        if (albumStatVoData == null) {
//            throw new ShisanException(201, "远程调用专辑微服务，获取专辑统计信息失败");
//        }
//        Integer playStatNum = albumStatVoData.getPlayStatNum();
//        Integer subscribeStatNum = albumStatVoData.getSubscribeStatNum();
//        Integer buyStatNum = albumStatVoData.getBuyStatNum();
//        Integer commentStatNum = albumStatVoData.getCommentStatNum();
//        albumInfoIndex.setPlayStatNum(playStatNum); // 专辑播放量
//        albumInfoIndex.setSubscribeStatNum(subscribeStatNum); // 专辑订阅量
//        albumInfoIndex.setBuyStatNum(buyStatNum); // 专辑购买量
//        albumInfoIndex.setCommentStatNum(commentStatNum); // 专辑评论量
//        // Double hotScore = commentStatNum * 0.1 + subscribeStatNum * 0.2 + buyStatNum * 0.3 + playStatNum * 0.4;
//        // 此处使用一个简单的算法计算热度值
//        Double hotScore = new Random().nextDouble();
//        albumInfoIndex.setHotScore(hotScore); // 专辑热度值
//
//        long endTime = System.currentTimeMillis();
//        log.info("专辑上架，耗时：{}毫秒", endTime - startTime);
//
//        // 将文档对象保存到es中
//        albumInfoIndexRepository.save(albumInfoIndex);
//    }
}
